package com.wish.g01.system.kafka.consument;

import com.alibaba.fastjson.JSON;
import com.wish.g01.common.redis.service.RedisService;
import com.wish.g01.system.api.domain.SysOperLog;
import com.wish.g01.system.service.ISysOperLogService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.listener.AcknowledgingMessageListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * 这个类的作者是13536
 * 开发日期:2022/5/24
 **/
@Component
@Slf4j
public class Listener3 implements AcknowledgingMessageListener<String,String> {

    @Autowired
    RedisService redisService;

    @Autowired
    ISysOperLogService iSysOperLogService;



    @Override
    @KafkaListener(topicPartitions = @TopicPartition(topic = "log",partitions = {"3"}))
    public void onMessage(ConsumerRecord<String, String> consumerRecord, Acknowledgment acknowledgment) {

        String key = consumerRecord.key();
        String value = consumerRecord.value();
        SysOperLog sysOperLog = JSON.parseObject(value, SysOperLog.class);
        log.info("token"+sysOperLog.getToken());
        if(!redisService.setCache(key)){
            return;
        }
        iSysOperLogService.insertOperlog(sysOperLog);

        log.info("kafka消费三区---");
        acknowledgment.acknowledge();
    }
}
